#pragma once

#include <butil/string_printf.h>

#include "index/attribute_index/attribute_disk_indexer/attribute_disk_indexer.h"

namespace hawking {
namespace indexlib {

DECLARE_int32(single_value_attribute_increase_initsize);

template<class T>
class SingleValueAttributeDiskIndexer : public AttributeDiskIndexer {
public:
    SingleValueAttributeDiskIndexer(
        int32_t field_id,
        util::ColumnType field_type,
        std::shared_ptr<AttributeConvertor> attribute_convertor) : 
            AttributeDiskIndexer(field_id, field_type, attribute_convertor) {}
    virtual ~SingleValueAttributeDiskIndexer() = default;

    virtual Status BatchRead(SegmentContext* segment_ctx) override;

    virtual uint32_t ElementLength() const override {
        return sizeof(T);
    }

    virtual uint64_t DocCount() const override {
        return data_reader_->Length() / sizeof(T);
    }

    virtual off_t GetOffset(DocId doc_id) const override {
        return doc_id * sizeof(T);
    }

    virtual bool IsVarlen() const {
        return false;
    }

protected:
    virtual util::HawkingRecallRetCode UpdateField_(
        DocId doc_id, const std::string_view& value) override;

private:
    void IncreaseRead_(
        doc_context* doc_ctx, T* updated_data);
    void MakeFilePart_(
        doc_context* doc_ctx,
        size_t offset_in_datafile,
        std::vector<FilePart>* immutable_datafile_part);
    void CopyDataToProtobuffer_(SegmentContext* segment_ctx);

    std::unique_ptr<HashMap<DocId, T>> update_dict_;
};

template<class T>
util::HawkingRecallRetCode SingleValueAttributeDiskIndexer<T>::UpdateField_(
    DocId doc_id, const std::string_view& value) {
    if (!update_dict_) {
        update_dict_ = std::make_unique<HashMap<DocId, T>(
            FLAGS_single_value_attribute_increase_initsize);
    }

    T new_value = *(const T*)(value.data());
    update_dict_->FindAndInsert(doc_id, new_value);
    return util::HawkingRecallRetCode::OK;
}

template<class T>
void SingleValueAttributeDiskIndexer<T>::IncreaseRead_(
    doc_context* doc_ctx, T* updated_data) {
    if (field_type_ == util::ColumnType::COLUMN_INT8 ||
        field_type_ == util::ColumnType::COLUMN_INT16 ||
        field_type_ == util::ColumnType::COLUMN_INT32 ||
        field_type_ == util::ColumnType::COLUMN_INT64) {
        doc_ctx->single_int_field.emplace(field_id_, *updated_data);
    } else if (field_type_ == util::ColumnType::COLUMN_FLOAT) {
        doc_ctx->single_float_field.emplace(field_id_, *updated_data);
    } else if (field_type_ == util::ColumnType::COLUMN_DOUBLE) {
        doc_ctx->single_double_field.emplace(field_id_, *updated_data);
    }
}

template<class T>
void SingleValueAttributeDiskIndexer<T>::MakeFilePart_(
    doc_context* doc_ctx,
    size_t offset_in_datafile,
    std::vector<FilePart>* immutable_datafile_part) {
    if (field_type_ == util::ColumnType::COLUMN_INT8 ||
        field_type_ == util::ColumnType::COLUMN_INT16 ||
        field_type_ == util::ColumnType::COLUMN_INT32 ||
        field_type_ == util::ColumnType::COLUMN_INT64) {
        doc_ctx->single_int_field.emplace(field_id_, 0);
        uint8_t* addr = reinterpret_cast<uint8_t*>(
            &(doc_ctx.single_int_field[field_id_]));
        immutable_datafile_part->emplace_back(offset_in_datafile, ElementLength(), addr);
    } else if (field_type_ == util::ColumnType::COLUMN_FLOAT) {
        doc_ctx->single_float_field.emplace(field_id_, 0);
        uint8_t* addr = reinterpret_cast<uint8_t*>(
            &(doc_ctx.single_float_field[field_id_]));
        immutable_datafile_part->emplace_back(offset_in_datafile, ElementLength(), addr);
    } else if (field_type_ == util::ColumnType::COLUMN_DOUBLE) {
        doc_ctx->single_double_field.emplace(field_id_, 0);
        uint8_t* addr = reinterpret_cast<uint8_t*>(
            &(doc_ctx.single_double_field[field_id_]));
        immutable_datafile_part->emplace_back(offset_in_datafile, ElementLength(), addr);
    }
}

template<class T>
void SingleValueAttributeDiskIndexer<T>::CopyDataToProtobuffer_(SegmentContext* segment_ctx) {
    for (auto& doc_ctx : doc_ctxs) {
        if (doc_ctx.filtered) {
            continue;
        }

        switch (field_type) {
        case util::ColumnType::COLUMN_INT8:
        case util::ColumnType::COLUMN_INT16:
        case util::ColumnType::COLUMN_INT32:
        case util::ColumnType::COLUMN_INT64: {
            auto iter = doc_ctx.single_int_field.find(field_id_);
            if (iter != doc_ctx.single_int_field.end()) {
                util::FieldValue* field_value =
                    segment_ctx->ArenaCtx().CreateOnArena<util::FieldValue>();
                field_value->set_int_value(iter->second);
                doc_ctx.field2attr.emplace(field_id_, field_value);
            }
        }
        break;
        case util::ColumnType::COLUMN_FLOAT: {
            auto iter = doc_ctx.single_float_field.find(field_id_);
            if (iter != doc_ctx.single_float_field.end()) {
                util::FieldValue* field_value =
                    segment_ctx->ArenaCtx().CreateOnArena<util::FieldValue>();
                field_value->set_double_value(iter->second);
                doc_ctx.field2attr.emplace(field_id_, field_value);
            }
        }
        break;
        case util::ColumnType::COLUMN_DOUBLE: {
            auto iter = doc_ctx.single_double_field.find(field_id_);
            if (iter != doc_ctx.single_double_field.end()) {
                util::FieldValue* field_value =
                    segment_ctx->ArenaCtx().CreateOnArena<util::FieldValue>();
                field_value->set_double_value(iter->second);
                doc_ctx.field2attr.emplace(field_id_, field_value);
            }
        }
        break;
        default:
            break;
        }
    }
}

template<class T>
Status SingleValueAttributeDiskIndexer<T>::BatchRead(SegmentContext* segment_ctx) {
    std::vector<FilePart> immutable_datafile_part;
    immutable_datafile_part.reserve(doc_context.size());
    for (size_t idx = 0; idx < doc_context.size(); ++idx) {
        if (doc_context[idx].filtered) {
            continue;
        }

        DocContext* doc_ctx = &doc_context[idx];
        T* updated_data = update_dict_.Find(segment_ctx->DocIds()[idx]);
        if (updated_data) {
            IncreaseRead_(doc_ctx, updated_data);
        } else {
            MakeFilePart_(
                doc_ctx,
                segment_ctx->DocIds()[idx] * ElementLength(),
                &immutable_datafile_part);
        }
    }

    if (!immutable_datafile_part.empty() &&
        (ret = data_reader_->BatchRead(immutable_datafile_part).Code()) != Status::OK) {
        return ret;
    }

    CopyDataToProtobuffer_(segment_ctx);
    return Status::OK;
}

}
}